package com.codeduck.kafka.quarzt;

import com.alibaba.fastjson.JSON;
import com.codeduck.kafka.kafka.KafkaInvoker;
import com.codeduck.kafka.pojo.TopicHandler;
import org.omg.CORBA.PRIVATE_MEMBER;
import org.omg.CORBA.TIMEOUT;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.*;

/**
 * @auther Mahone Yan
 * @date 2023/5/7
 */
@Component
public class ServerLogProducerJob {
    private static final Logger LOGGER = LoggerFactory.getLogger(AcctDetailProducerJob.class);
    private static final Integer POOL_SIZE = 10000;
    private static final Integer KEEP_ALIVE_TIME = 6000;
    private ThreadPoolExecutor executor;
    private static final String[] charArr = new String[]{"a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k", "l", "m", "n", "i", "o", "p", "q", "r", "s", "t", "u", "v", "w", "x", "y", "z",
            "A", "B", "C", "D", "E", "F", "G", "H", "I", "J", "K", "L", "M", "N", "O", "P", "Q", "R", "S", "T", "U", "V", "W", "X", "Y", "Z"};
    @Resource
    private KafkaInvoker kafkaInvoker;

//    @PostConstruct
    public void init() {
        this.run();
    }
    public void run() {
        int cpuNum = Runtime.getRuntime().availableProcessors();
        try {
            executor = new ThreadPoolExecutor(1, 1, KEEP_ALIVE_TIME,
                    TimeUnit.MILLISECONDS,  new LinkedBlockingDeque<>(POOL_SIZE));
            while (true) {
                Future<Integer> future = executor.submit(doProduceLog());
                LOGGER.info("已推送日志数据[{}]条至Kafka", future.get());
                TimeUnit.MILLISECONDS.sleep(1000);
            }
        } catch (Exception e) {
            LOGGER.info("批量日志上送Kafka出现异常", e);
        }
    }

    private Callable<Integer> doProduceLog() {
        Callable<Integer> callable = () -> {
            List<AppLog> logs = getAppLogs(new Random().nextInt(1000));
            kafkaInvoker.sendMessage(JSON.toJSONString(logs), TopicHandler.SERVICE_LOG_TOPIC.getTopic());
            return logs.size();
        };
        return callable;
    }

    private List<AppLog> getAppLogs(int count) {
        List<AppLog> logs = new ArrayList<>(count);
        for (int i = 0; i < count; i++) {
            AppLog appLog = new AppLog(System.currentTimeMillis(), this.getClass().getName(), "AppLog_" + getLogInfo());
            logs.add(appLog);
        }
        return logs;
    }

    private String getLogInfo() {
        StringBuilder sb = new StringBuilder();
        Random random = new Random();
        for (int i = 0; i < random.nextInt(100); i++) {
            sb.append(charArr[random.nextInt(charArr.length)]);
        }
        return sb.toString();
    }


    static class AppLog {
        private long timestamp;
        private String className;
        private String log;

        public AppLog(long timestamp, String className, String log) {
            this.timestamp = timestamp;
            this.className = className;
            this.log = log;
        }

        public long getTimestamp() {
            return timestamp;
        }

        public String getClassName() {
            return className;
        }

        public String getLog() {
            return log;
        }
    }

}
